package kuoge.practice.kafka.stream.member.clickrate;

import com.alibaba.fastjson.JSON;
import io.netty.util.HashedWheelTimer;
import io.netty.util.TimerTask;
import kuoge.practice.kafka.stream.member.KafkaProducerFactory;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.concurrent.TimeUnit;

import static kuoge.practice.kafka.stream.member.TopicConstants.ITEM_CLICK_TOPIC;

/**
 * @Description
 */
public class UserClickEventProducer {
    private static final Producer<String, String> kafkaProducer = KafkaProducerFactory.getInstance();
    private final HashedWheelTimer hashedWheelTimer = new HashedWheelTimer();

    void startSendEvent() {
        user_a_event();

        user_b_event();

        System.out.println("start send event success");
    }

    private void user_a_event() {
        TimerTask task = timeout -> {
            final UserClickEvent userClickEvent = new UserClickEvent("member_a", "xian_rou_da_bao");
            kafkaProducer.send(new ProducerRecord<>(ITEM_CLICK_TOPIC, userClickEvent.getItemCode(), JSON.toJSONString(userClickEvent)), (rs, ex) -> System.out.println(rs.offset() + ", " + ex));
            System.out.println("send member_a client event, thread:" + Thread.currentThread().getName());
            user_a_event();
        };
        hashedWheelTimer.newTimeout(task, 1000, TimeUnit.MILLISECONDS);
    }

    private void user_b_event() {
        TimerTask task = timeout -> {
            final UserClickEvent userClickEvent = new UserClickEvent("member_b", "nian_huo_la_rou");
            kafkaProducer.send(new ProducerRecord<>(ITEM_CLICK_TOPIC, userClickEvent.getItemCode(), JSON.toJSONString(userClickEvent)), (rs, ex) -> System.out.println(rs.offset() + ", " + ex));
            System.out.println("send member_b click event, thread:" + Thread.currentThread().getName());
            user_b_event();
        };
        hashedWheelTimer.newTimeout(task, 3000, TimeUnit.MILLISECONDS);
    }

    void stopSendEvent() {
        hashedWheelTimer.stop();
    }
}
